package com.hhxy.rabbitmq.demo04;

import com.hhxy.rabbitmq.utils.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ConcurrentSkipListSet;

/**
 * @author ghp
 * @date 2023/2/23
 * @title
 * @description
 */
public class ConfirmMessage {

    // 消息的生产数量
    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
//        publishMessageIndividually(); // 单个发布确认
//        publishMessageBatch(); // 批量发布确认
        publishMessageAsync(); // 异步批量确认
    }

    /**
     * 单个发布确认
     */
    public static void publishMessageIndividually() throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        String queueName = UUID.randomUUID().toString();
        /*
            1. 队列名称
            2. 队列中的消息是否持久化（磁盘），默认取值为false，表示不持久化，此时消息存储在内存中
            3. 队列是否排他，true表示只能同一个连接中的信道使用，false表示不同连接的信道都可以使用该队列
            4. 是否自动删除，true表示当所有消费者与该队列断开了连接，队列会自动删除
            5. 其它参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        // 开启发布确认
        channel.confirmSelect();

        long startTime = System.currentTimeMillis(); // 发送消息的开始时间

        // 批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            /*
               1. 指定要发送的交换机，空表示使用默认的交换机
               2. 指定将消息存放到哪一个队列
               3. 其它参数
               4. 指定本次要发送的消息
            */
            channel.basicPublish("", queueName, null, message.getBytes());
            // 获取确认消息(true表示消息写入磁盘成功，false表示写入磁盘失败)
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息持久化成功！");
            }
        }

        long endTime = System.currentTimeMillis(); // 发送消息的结束时间

        System.out.println("发布" + MESSAGE_COUNT + "条消息耗时: " + (endTime - startTime) + "ms");

    }

    /**
     * 批量发布确认
     */
    public static void publishMessageBatch() throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        String queueName = UUID.randomUUID().toString();
        /*
            1. 队列名称
            2. 队列中的消息是否持久化（磁盘），默认取值为false，表示不持久化，此时消息存储在内存中
            3. 队列是否排他，true表示只能同一个连接中的信道使用，false表示不同连接的信道都可以使用该队列
            4. 是否自动删除，true表示当所有消费者与该队列断开了连接，队列会自动删除
            5. 其它参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        // 开启发布确认
        channel.confirmSelect();

        // 批量确认消息的条数，发送100条消息后再确认消息是否写入磁盘
        int batchSize = 100;
        int unconfirmed; // 确认消息的条数

        long startTime = System.currentTimeMillis(); // 发送消息的开始时间

        // 批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            /*
               1. 指定要发送的交换机，空表示使用默认的交换机
               2. 指定将消息存放到哪一个队列
               3. 其它参数
               4. 指定本次要发送的消息
            */
            channel.basicPublish("", queueName, null, message.getBytes());
            // 每发送100条消息，就确认一次
            if (i % batchSize == 0) {
                // 获取确认消息(true表示消息写入磁盘成功，false表示写入磁盘失败)
                boolean flag = channel.waitForConfirms();
                if (flag) {
                    System.out.println("消息持久化成功！");
                }
            }
        }

        long endTime = System.currentTimeMillis(); // 发送消息的结束时间

        System.out.println("发布" + MESSAGE_COUNT + "同步条消息耗时: " + (endTime - startTime) + "ms");
    }

    /**
     * 异步批量确认
     */
    public static void publishMessageAsync() throws Exception {
        Channel channel = RabbitMqUtil.getChannel();
        String queueName = UUID.randomUUID().toString();
        /*
            1. 队列名称
            2. 队列中的消息是否持久化（磁盘），默认取值为false，表示不持久化，此时消息存储在内存中
            3. 队列是否排他，true表示只能同一个连接中的信道使用，false表示不同连接的信道都可以使用该队列
            4. 是否自动删除，true表示当所有消费者与该队列断开了连接，队列会自动删除
            5. 其它参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        // 开启发布确认
        channel.confirmSelect();

        /*
            线程安全且有序的哈希表，用于存储发送的消息
            1. 能轻松将序号和消息进行关联
            2. 能够轻松根据序号批量删除消息
            3. 支持高并发
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms =
                new ConcurrentSkipListMap<>();


        long startTime = System.currentTimeMillis(); // 发送消息的开始时间

        // 开启消息监听器，负责监听哪些消息发送失败，哪些消息发送成功
        /*
            1. 发送成功消息的标识
            2. 是否批量确认
         */
        ConfirmCallback ackCallBack = (deliveryTag, multiple) -> {
            // 每发送一条成功的消息，就执行这个回调中的逻辑
            // 删除已确认的消息
            if(multiple){
                // 如果是批量确认的，就清空hash表
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag);
                confirmed.clear();
            }else{
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("消息写入磁盘成功:" + deliveryTag);
        };
        ConfirmCallback nackCallBack = (deliveryTag, multiple) -> {
            // 每发送一条失败的消息，就执行这个回调中的逻辑
            String message = outstandingConfirms.get(deliveryTag);
            System.out.println("未确认的消息:" +message+ "   未确认消息的tag:"+ deliveryTag);
        };
        /*
            1. 监听发送成功的消息
            2. 监听发送失败的消息
         */
        channel.addConfirmListener(ackCallBack, nackCallBack);

        // 异步批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            /*
               1. 指定要发送的交换机，空表示使用默认的交换机
               2. 指定将消息存放到哪一个队列
               3. 其它参数
               4. 指定本次要发送的消息
            */
            channel.basicPublish("", queueName, null, message.getBytes());
            // 记录所有发送的消息
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
        }

        long endTime = System.currentTimeMillis(); // 发送消息的结束时间

        System.out.println("发布" + MESSAGE_COUNT + "条异步消息耗时: " + (endTime - startTime) + "ms");
    }
}
